package com.lagou.rocket.demo.order.consumer;

import com.lagou.rocket.demo.order.model.Inventory;
import com.lagou.rocket.demo.order.model.Order;
import com.lagou.rocket.demo.order.service.IInventoryService;
import com.lagou.rocket.demo.order.service.IOrderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;


/**
 * <p>
 *
 * </p>
 *
 * @author KK
 * @since 2021-11-07
 */
@Slf4j
@Component
@RocketMQMessageListener(topic = "seckill", consumerGroup = "seckill_listener")
public class SecKillListener implements RocketMQListener<Order> {


    private final RocketMQTemplate rocketMQTemplate;
    private final IOrderService orderService;
    private final IInventoryService inventoryService;

    private final RedisTemplate redisTemplate;

    // 库存预检查key的前缀
    private final static String PREFIX = "inventory_";


    private final PlatformTransactionManager transactionManager;




    public SecKillListener(RocketMQTemplate rocketMQTemplate, IOrderService orderService, IInventoryService inventoryService, RedisTemplate redisTemplate, PlatformTransactionManager transactionManager) {
        this.rocketMQTemplate = rocketMQTemplate;
        this.orderService = orderService;
        this.inventoryService = inventoryService;
        this.redisTemplate = redisTemplate;
        this.transactionManager = transactionManager;
    }


    @Override
    public void onMessage(Order message) {
        // 这个消息的消费应该是幂等的
        log.info("消费下单消息， 消息内容：{}", message);


        log.info("发送订单延迟消息， 级别为4， 1分钟后消费");

        //  发送一条延迟检查订单的支付状态的消息; 这个消息是事务的
        Message checkPayMsg = MessageBuilder
                .withPayload(message.getId())
                .build();


        SendResult checkPaySendResult = rocketMQTemplate.syncSend(
                "check_pay",
                checkPayMsg,
                10 * 1000L,
                4
        );
        if (!SendStatus.SEND_OK.equals(checkPaySendResult.getSendStatus())){
            log.info("发送延迟消息失败");
            throw new RuntimeException("发送延迟消息失败");
        }


        boolean decreased = false;

        // 开始数据库事务
        TransactionDefinition transactionDefinition = new DefaultTransactionDefinition();
        log.info("开始rocketMq + 本地mysql事务........");
        TransactionStatus transactionStatus = transactionManager.getTransaction(transactionDefinition);



        String productCode = message.getProductCode();

        //查库存(select ... in share mode),
        Inventory inventory = null;
        try {
            log.info("查询实际库存， 同时对库存上锁........");
            inventory = inventoryService.getInventoryByProductCode(productCode);
            log.info("库存状态：{}", inventory);


            //如果库存存在:
            if (inventory.getInventory() > 0) {
                //  创建订单;
                message.setStatus("unpaid");

                log.info("保存订单.....");

                //  订单写入数据库;
                orderService.save(message);

                //  数据库中减库存;
                inventory.setInventory(inventory.getInventory() - 1);
                inventoryService.save(inventory);

                decreased = true;



            } else {
                log.info("订单售罄.....");
                log.info("保存为售罄订单.....");
                //  创建一条售罄的订单;
                message.setStatus("sell_out");
                //   写入数据库;
                orderService.save(message);
                log.info("提交本地mysql事务");
            }

            transactionManager.commit(transactionStatus);
        } catch (Exception e) {
            log.error("出现异常， 回滚本地事务。 ", e);
            transactionManager.rollback(transactionStatus);
            return;
        }

        //扣减redis中的缓存;
        if (decreased){
            log.info("扣减redis中的缓存.....");
            String inventoryKey = PREFIX + message.getProductCode();
            Long decrement = redisTemplate.opsForValue().decrement(inventoryKey);
            log.info("扣减之后的库存为： {}", decrement);
        }

    }
}
